package com.dongdongshop.consume;

import cn.hutool.core.util.RandomUtil;
import com.aliyun.dysmsapi20170525.Client;
import com.aliyun.dysmsapi20170525.models.SendSmsRequest;
import com.aliyun.tea.TeaException;
import com.aliyun.teaopenapi.models.Config;
import com.aliyun.teautil.models.RuntimeOptions;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.TimeUnit;

@Component
@RocketMQMessageListener(topic = "Phone-Message-1",consumerGroup = "my-group-consumer-topic-1")
public class PhoneMesConsume implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {

    @Autowired
    private RedisTemplate redisTemplate;
    @Override
    public void onMessage(String s) {
        //方法作废
    }
    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        //手动接管ACK机制,如果消费者失败,mq会重试:默认16次,时隔越来越长
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt message = list.get(0);
                String msgId = message.getMsgId();//消息的ID
                int reconsumeTimes = message.getReconsumeTimes();//消息重试次数
                System.out.println("消息的id:"+msgId);
                System.out.println("消息的重试次数:"+reconsumeTimes);
                String phone = new String(message.getBody());//获取消息实体,这个时候就不支持直接接受对象
                try{
                    System.out.println("拿到消息内容:"+phone);
                    //消费之前查看缓存中有没有消费过
                    Boolean mqmessage = redisTemplate.boundHashOps("message").hasKey(phone);
                    if (!mqmessage){
                        System.out.println("消费我们的消息:"+phone);

                        Client client = createClient("LTAI5t8egQ8mrPYi5LLEcNY7", "XtLOW5rpVbgwIGHEuCCh95sFVP2gUa");
                        String code = RandomUtil.randomNumbers(4);
                        SendSmsRequest sendSmsRequest = new SendSmsRequest()
                                .setPhoneNumbers(phone)
                                .setSignName("东科创想")
                                .setTemplateCode("SMS_162522027")
                                .setTemplateParam("{\"sms_code\":\"" + code + "\"}");
                        RuntimeOptions runtime = new RuntimeOptions();
                        try {
                            // 复制代码运行请自行打印 API 的返回值
                            client.sendSmsWithOptions(sendSmsRequest, runtime);
                        } catch (TeaException error) {
                            // 如有需要，请打印 error
                            com.aliyun.teautil.Common.assertAsString(error.message);
                        } catch (Exception _error) {
                            TeaException error = new TeaException(_error.getMessage(), _error);
                            // 如有需要，请打印 error
                            com.aliyun.teautil.Common.assertAsString(error.message);
                        }

//                        int i=9/0;
                        redisTemplate.boundHashOps("message").put(phone,code);//将这条数据存入redis中
                        redisTemplate.expire("message",30, TimeUnit.MINUTES);//设置30分钟的过期时间
                    }else {
                        System.out.println("该消息已经消费过了,将不再消费");
                    }
                }catch (Exception e){
                    e.printStackTrace();
                    if (reconsumeTimes >= 3){
                        System.out.println("重试次数过多,并且消息仍然消费失败,那么我们就将消息存在死信队列");
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//手动结束,告诉这条消息别再重试了
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }

    /**
     * 使用AK&SK初始化账号Client
     * @param accessKeyId
     * @param accessKeySecret
     * @return Client
     * @throws Exception
     */
    public  com.aliyun.dysmsapi20170525.Client createClient(String accessKeyId, String accessKeySecret) throws Exception {
        Config config = new Config()
                // 您的 AccessKey ID
                .setAccessKeyId(accessKeyId)
                // 您的 AccessKey Secret
                .setAccessKeySecret(accessKeySecret);
        // 访问的域名
        config.endpoint = "dysmsapi.aliyuncs.com";
        return new com.aliyun.dysmsapi20170525.Client(config);
    }
}
